#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2023/5/28 23:04
# @Author  : huidong.bai
# @File    : assertion.py
# @Software: PyCharm
# @Mail    : MasterBai2018@outlook.com

import pdb
import os
import re
import json
import allure
import fnmatch
from queue import Queue
from threading import Event
from conftest import logging
from src.core.Case import check_fields
from src.core.Case import TestCase
from src.core.Common.ResultStorage import ResultStorage
from concurrent.futures import ThreadPoolExecutor, as_completed
from src.core.Common.Bean.RuntimeRecorder import RuntimeRecorder
from src.core.Common.Bean.XMLReporter import MetaErrorInfo, XMLReporter
from src.core.Status.AIBSSessionStatus import MongoCode
from src.utils.common import async_color_print, load_yaml_config


class AssertionFactory:
    def __init__(self):
        self.resultStorage = None
        self.assertion_results = None
        self.dataBus = None
        self.recoder = None
        self.case = None
        self.result_path = None
        self.reporter = None
        self.collect_data = []
        self.json_data_list = None
        self.expect_results = []
        self.actual_results = []
        self.isPass = ''
        self.parse_yaml = None
        self.result = False
        self.yamlKeys = None
        self.assert_finish_cond = None
        self.actual_data = None
        self.expect_data = None
        self.tts_data = None
        self.database = None
        self.project = None

    def set_case(self, case: TestCase):
        self.case = case

    def set_finish_event(self, assert_event: Event):
        self.assert_finish_cond = assert_event

    def set_project(self, project):
        self.project = project
    
    def setResultPath(self, result_path):
        self.result_path = result_path
        with open(self.result_path, "w+", encoding="utf-8", buffering=-1) as fp:
            fp.write("VoiceInputPath\t\t\tExpectResults\t\t\tActualResults\t\t\tisPass\n")

    def load_path_yaml(self, yaml_config):
        keys = set()
        self.yamlKeys = yaml_config.keys()
        for section in yaml_config:
            keys |= set(yaml_config[section].keys())
        keys |= set(RuntimeRecorder().__dict__.keys())
        self.parse_yaml = keys

    def set_reporter(self, reporter: XMLReporter):
        self.reporter = reporter

    def set_recorder(self, recoder: RuntimeRecorder):
        self.recoder = recoder

    def set_databus(self, dataBus: Queue):
        self.dataBus = dataBus
    
    def set_database(self, database):
        self.database = database

    def assert_results(self):
        line_pass = []
        # 循环获取数据通道的callback数据
        while True:
            callback_data = self.dataBus.get()
            msg_status = callback_data.get("status")
            if msg_status not in [MongoCode.STATUS_MONGO_ASSERT, MongoCode.STATUS_MONGO_FINISH]:
                # 收集断言结果
                self.collect_data.append(callback_data)
                continue
            # 过滤断言预期
            except_list = [item for item in self.case.expect if item.get("voiceRegion")[1] == callback_data.get("voiceRegionEnd")]
            for item in except_list:
                self.case.expect.remove(item)

            if msg_status == MongoCode.STATUS_MONGO_ASSERT:
                for exp in except_list:
                    channel = exp.get("channel")
                    msgType = exp.get("msgType")
                    exp_data = exp.get("expects")
                    if msgType == "API":
                        line_pass.append(self.assert_line_info(self.recoder.__dict__, exp_data, msgType, channel))
                    else:
                        result = self.filter_result(channel, msgType)
                        line_pass.append(self.assert_line_info(result, exp_data, msgType, channel))
                self.collect_data.clear()
                continue
            if msg_status == MongoCode.STATUS_MONGO_FINISH:
                for exp in except_list:
                    channel = exp.get("channel")
                    msgType = exp.get("msgType")
                    exp_data = exp.get("expects")
                    if msgType == "API":
                        line_pass.append(self.assert_line_info(self.recoder.__dict__, exp_data, msgType, channel))
                    else:
                        result = self.filter_result(channel, msgType)
                        line_pass.append(self.assert_line_info(result, exp_data, msgType, channel))
                self.collect_data.clear()
                break
        self.result = all(line_pass)

    def filter_result(self, channel, message_type):
        for item in self.collect_data:
            result_channel = item.get('channelId')
            if item['resultType'] == message_type and str(result_channel) == str(channel):
                filter_msg = item
                self.collect_data.remove(filter_msg)
                return filter_msg

    def routine_assert_results(self, resultStorage, expect_type, expect_result, channel_id=-1):
        self.resultStorage = resultStorage
        actual_result = self.select_result(channel_id, expect_type)
        pass_tag = self.assert_line_info(actual_result, expect_result, channel_id, expect_type)

        # 将结果写入报告中
        if not pass_tag:
            inputs = "RoutineCaseInput"
            case_name = "RoutineCase"
            error_info = MetaErrorInfo(inputs, ";".join(self.expect_results), ";".join(self.actual_results))
            self.reporter.push_result(case_name, error_info)
            self.assertion_results.append(pass_tag)

        self.print_msg(pass_tag, expect_type, channel_id)
        self.actual_results.clear()
        self.expect_results.clear()

    def assert_line_info(self, result, expect_info: str, assert_type, channel):
        line_result_list = []
        self.isPass = ''
        assert_dict = check_fields(expect_info, self.parse_yaml)

        ttsID = assert_dict.get("ttsID")
        if ttsID is not None:
            self.tts_data = self.database.query_data(table=f'TTS_{self.project}', condition="id", condition_value=ttsID)

        for key in assert_dict.keys():
            expect_value = assert_dict.get(key)
            self.expect_data = expect_value if expect_value not in [None, "None", ""] else ""
            if result is None:
                meta_pass = self.assert_meta(key, expect_value, "None")
                self.actual_data = ""
            else:
                actual_value = result.get(key)
                self.actual_data = actual_value if actual_value not in [None, "None", ""] else ""
                meta_pass = self.assert_meta(key, expect_value, actual_value)
            line_result_list.append(meta_pass)
        line_result = all(line_result_list)
        self.print_msg(line_result, assert_type, channel)
        return line_result

    def select_result(self, channel, message_type):
        if channel == 0:
            for item in self.resultStorage:
                if item['resultType'] == message_type:
                    filter_msg = item
                    self.resultStorage.remove(filter_msg)
                    return filter_msg
        else:
            for item in self.resultStorage:
                result_channel = item.get('channelId')
                if item['resultType'] == message_type and str(result_channel) == str(channel):
                    filter_msg = item
                    self.resultStorage.remove(filter_msg)
                    return filter_msg

    def assert_meta(self, key, expect_value: str, actual_value: str):
        # 当断言值为voiceRegion时,跳过断言,不进行处理
        if key == "voiceRegion" or key == "ttsID":
            self.expect_results.append(f"{key}:{expect_value}")
            return True
        if key == "tts":
            if self.tts_data is not None and len(self.tts_data.strip()) > 0:
                return self._fuzz_equal_tts_data(actual_value)
            else:
                return self._fuzz_equal(key, expect_value, actual_value)
        if "*" in expect_value:
            return self._fuzz_equal(key, expect_value, actual_value)
        if "±" in expect_value:
            return self.tolerance_equal(key, expect_value, actual_value)
        else:
            return self._equal(key, expect_value, actual_value)
    
    def _fuzz_equal_tts_data(self, actual):
        flag = False
        # 模糊比较
        if not str(actual) or actual == "None":
            if self.tts_data is None or self.tts_data == "None":
                actual = "None"
                flag = True
            else:
                flag = False
        else:
            actual = str(actual)
            for expect in self.tts_data.split("|"):
                pattern = re.escape(expect).replace(r'\*', r'.*')
                match = re.match(pattern, actual)
                flag = match is not None and len(match.group()) == len(actual)
                if flag:
                    break

        self.expect_results.append(f"tts:{self.tts_data}")
        self.actual_results.append(f"tts:{actual}")
        return flag

    def _fuzz_equal(self, key, expect, actual):
        # 模糊比较
        if not str(actual) or actual == "None":
            if expect == "None":
                actual = expect
                flag = True
            else:
                flag = False
        elif (expect is not None) and ("|" in expect):
            flag = False
            part_list = expect.split("|")
            for item in part_list:
                pattern = re.escape(item).replace(r'\*', r'.*')
                match = re.match(pattern, str(actual))
                _tag = match is not None and len(match.group()) == len(str(actual))
                if _tag == True:
                    flag = True
        else:
            actual = str(actual)
            pattern = re.escape(expect).replace(r'\*', r'.*')
            match = re.match(pattern, actual)
            flag = match is not None and len(match.group()) == len(actual)

        self.expect_results.append(f"{key}:{expect}")
        self.actual_results.append(f"{key}:{actual}")
        return flag

    def tolerance_equal(self, key, expect, actual):
        if expect == "None":
            expect = None
        if actual == "None":
            actual = None
        if not actual and actual != 0:
            _tag = expect == actual
            self.expect_results.append(f"{key}:{expect}")
            self.actual_results.append(f"{key}:{actual}")
        else:
            expect = "0" + expect
            tag, tolerance = [int(i) for i in expect.split("±")]
            tag_start, tag_end = [tag - tolerance, tag + tolerance]
            _tag = tag_start <= int(actual) <= tag_end
            self.expect_results.append(f"{key}:{expect[1:]}")
            self.actual_results.append(f"{key}:{actual}")
        return _tag

    def _abso_equal_json(self, keys, actual, expect):
        # json完全匹配
        for key, value in expect.items():
            currentKey = keys + '.' + key
            if key not in actual:
                return expect
            else:
                if isinstance(value, dict):
                    self._abso_equal_json(currentKey, actual[key], value)
                elif isinstance(value, list):
                    if isinstance(actual[key], list):
                        for l, val in enumerate(value):
                            for ac_val in actual[key]:
                                self._abso_equal_json(currentKey, ac_val, val)
                else:
                    if value == "*":
                        expect[key] = actual[key]
        return expect

    def _fuzz_equal_json(self, keys, actual, expect, flag=''):
        # json模糊匹配
        for key, value in expect.items():
            currentKey = keys + '.' + key
            if key not in actual:
                if value == "No-Struct":  # 预期字段不存在，实际字段不存在
                    self.isPass += '1'
                else:
                    self.isPass += '0'
                    logging.error(f"[ ERROR ] Key 【 {currentKey} 】 not found in actual result")
                    async_color_print((f"[ ERROR ] Key 【 {currentKey} 】 not found in actual result", 'red'))
            else:
                if isinstance(value, dict):
                    _tag = self._fuzz_equal_json(currentKey, actual[key], value)
                    if _tag:
                        self.isPass += '1'
                    else:
                        self.isPass += '0'
                # 若为数组，不根据顺序判断，但是：同时断言数组内json不存在某结构与存在某结构时会出问题，比如：[{"type": "No-Struct", "data": "No-Struct"},{"type": "AppDIR.open", "data": {"page": "default"}}]
                elif isinstance(value, list):
                    if isinstance(actual[key], list):
                        for val in value:
                            for i, ac_val in enumerate(actual[key]):
                                _tag = self._fuzz_equal_json(currentKey, ac_val, val, 'array')
                                if not _tag and i < len(actual[key]) - 1:
                                    continue
                                elif not _tag:
                                    self.isPass += '0'
                                    logging.error(
                                        f"[ ERROR ] Key 【 {currentKey} 】 Value 【 {val} 】 This content not found in actual result")
                                    async_color_print((
                                                      f"[ ERROR ] Key 【 {currentKey} 】 Value 【 {val} 】 This content not found in actual result",
                                                      'red'))
                                else:
                                    self.isPass += '1'
                                    actual[key].remove(ac_val)
                                    break
                    else:
                        self.isPass += '0'
                        logging.error(f"[ ERROR ] Key 【 {currentKey} 】 Expected array type")
                        async_color_print((f"[ ERROR ] Key 【 {currentKey} 】 Expected array type", 'red'))
                else:
                    if value == "*":
                        self.isPass += '1'
                    elif isinstance(value, str) and '*' in value:
                        if fnmatch.fnmatch(actual[key], value):
                            self.isPass += '1'
                        else:
                            self.isPass += '0'
                    elif actual[key] == value:
                        self.isPass += '1'
                    elif value == "No-Struct":  # 预期字段不存在，但实际字段存在
                        self.isPass += '0'
                        logging.error(f"[ ERROR ] Key 【 {currentKey} 】 should not exist in the actual result")
                        async_color_print(
                            (f"[ ERROR ] Key 【 {currentKey} 】 should not exist in the actual result", 'red'))
                    elif flag == 'array':
                        return False
                    else:
                        self.isPass += '0'
                        logging.error(f"[ ERROR ] Content 【 {currentKey} : {value} 】 not found in actual result")
                        async_color_print(
                            (f"[ ERROR ] Content 【 {currentKey} : {value} 】 not found in actual result", 'red'))
        if self.isPass[-1] == '1':
            return True
        else:
            return False

    def _equal(self, key, expect, actual):
        if actual == "":
            actual = "None"
        if expect == "None":
            expect = None
        if expect == ['']:
            expect = []
        if not actual and actual != 0:
            _tag = expect == actual
        else:
            if (expect is not None) and isinstance(expect, str) and expect.startswith("!"):
                _tag = str(expect[1:]) != str(actual)
            elif (expect is not None) and ("|" in expect):
                _tag = str(actual) in expect.split("|")
            elif isinstance(expect, list) and isinstance(actual, list):
                _tag = set(expect) == set(actual)
            else:
                _tag = str(expect) == str(actual)

        self.expect_results.append(f"{key}:{expect}")
        self.actual_results.append(f"{key}:{actual}")
        return _tag

    def write_nlu_rate(self):
        skill_rate = self.result_path.replace(os.path.basename(self.result_path), "skill_rate_result.txt")
        intention_rate = self.result_path.replace(os.path.basename(self.result_path), "intention_rate_result.txt")
        tts_rate = self.result_path.replace(os.path.basename(self.result_path), "tts_rate_result.txt")
        slots_rate = self.result_path.replace(os.path.basename(self.result_path), "slots_rate_result.txt")
        for index in range(len(self.expect_results)):
            expect_data = self.expect_results[index]
            actual_data = self.actual_results[index]
            if "skill" in expect_data:
                with open(skill_rate, "a+", encoding="utf-8") as fp:
                    fp.write(f"{self.case.input}\t{expect_data}\t{actual_data}\t{expect_data==actual_data}\n")
            if "intention" in expect_data:
                with open(intention_rate, "a+", encoding="utf-8") as fp:
                    fp.write(f"{self.case.input}\t{expect_data}\t{actual_data}\t{expect_data==actual_data}\n")
            if "tts" in expect_data:
                with open(tts_rate, "a+", encoding="utf-8") as fp:
                    fp.write(f"{self.case.input}\t{expect_data}\t{actual_data}\t{self._fuzz_equal('tts', expect_data, actual_data)}\n")
            else:
                with open(slots_rate, "a+", encoding="utf-8") as fp:
                    fp.write(f"{self.case.input}\t{expect_data}\t{actual_data}\t{expect_data==actual_data}\n")


    def print_msg(self, is_pass: bool, assert_type, channel):
        expect_infos = ";".join(self.expect_results)
        actual_infos = ";".join(self.actual_results)
        
        self.write_nlu_rate()

        expect_msg = f"[ 预期 ] [{channel}][{assert_type}] " + expect_infos + "\n"
        actual_msg = f"[ 实际 ] [{channel}][{assert_type}] " + actual_infos

        self.actual_results.clear()
        self.expect_results.clear()

        try:
            with allure.step(f'{expect_msg}'):
                pass
            with allure.step(f'{actual_msg}'):
                if not is_pass:
                    raise AssertionError
        except Exception as e:
            pass

        if self.result_path and assert_type in ["NLPResult", "ASRResult"]:
            with open(self.result_path, "a+", encoding="utf-8", buffering=-1) as fp:
                fp.write(f"{self.case.input.get('voice')}\t\t{expect_infos}\t\t{actual_infos}\t\t{is_pass}\n")

        async_color_print(("-" * 60 + "\n", "cyan"), (expect_msg, None), (actual_msg, 'green' if is_pass else 'red'), ("\n" + "-" * 60, "cyan"))
        
        if not is_pass:
            inputs = ''
            params = ''
            for key, value in self.case.input.items():
                inputs += f"{key}:{value};"
            for key, value in self.case.param.items():
                params += f"{key}:{value};"
            case_name = f"{inputs}\t{params}"
            error_info = MetaErrorInfo(inputs, f"[{channel}][{assert_type}]" + expect_infos, f"[{channel}][{assert_type}]" + actual_infos)
            self.reporter.push_result(case_name, error_info)

    def reset(self):
        self.expect_results = []
        self.actual_results = []
        self.result = False
